package org.xydpeng.framework.l2cache.serialization;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.util.Pool;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;

/**
 * @author xydpeng
 * @version v1.0
 * @date: 2024/6/12
 * @description:
 */
public class KryoSerialization implements Serialization {

    private static final Pool<Kryo> kryoPool = new Pool<Kryo>(true, false, 16) {
        protected Kryo create() {
            Kryo kryo = new Kryo();
            kryo.register(org.xydpeng.framework.l2cache.message.CacheMessage.class);
            kryo.register(org.xydpeng.framework.l2cache.enums.CacheOperation.class);
            // Configure the Kryo instance.
            return kryo;
        }
    };

    @Override
    public <T> byte[] serialize(T obj) {
        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
             Output output = new Output(baos)) {
            Kryo kryo = kryoPool.obtain();
            try {
                kryo.writeClassAndObject(output, obj);
                output.flush();
                return baos.toByteArray();
            } finally {
                kryoPool.free(kryo);
            }
        } catch (Exception e) {
            throw new RuntimeException("Serialization failed", e);
        }
    }

    @Override
    public Object deserialize(byte[] bytes) {
        try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
             Input input = new Input(bais)) {
            Kryo kryo = kryoPool.obtain();
            try {
                return kryo.readClassAndObject(input);
            } finally {
                kryoPool.free(kryo);
            }
        } catch (Exception e) {
            throw new RuntimeException("Deserialization failed", e);
        }
    }
}
